#!/usr/bin/perl
use strict;
use XML::Simple;
use Getopt::Std;
#use Socket;
use IO::Socket::INET qw(:DEFAULT :crlf);
use Fcntl;
use IO::Epoll;
use URI::URL;
use HTTP::Request;
#use HTTP::Headers;
use HTTP::Response;
use IO::Uncompress::Gunzip qw(gunzip $GunzipError);
use Compress::Zlib;

use vars qw($opt_a);

getopts('c:');

my $conf_file = "crawl_page.xml";
$conf_file = $opt_a if $opt_a;

#$/ = CRLF . CRLF;

#读取配置xml的变量
my $conf;
#用于判断是否重读配置的标志变量
my $reload = 0;
#三个服务的主机地址和端口，用于负载均衡
my @url_dist_host;
my @page_collect_host;
my @dns_cache_host;
#三个服务的通信socket
my $url_dist_sock = undef;
my $page_collect_sock = undef;
my $dns_cache_sock = undef;
#对http client的配置
my $total_client_num;
my $user_agent;
my $accept;
#需要URL分发进程提供给这些host的URL, undef表示全部
my @whose_url = undef;

$SIG{HUP} = sub {$reload = 1;};

&init();
my $epfd = epoll_create(10);
#主机端口对应的IO::Socket::INET对象
my %name_sockobj;
#socket数字描述符对应的IO::Socket::INET对象
#$fd_sockobj{$fd}[0]为IO::Socket::INET对象,$fd_sockobj{$fd}[1]为fd对应的host,$fd_sockobj{$fd}[2]为fd对应的port
my %fd_sockobj;
#$loc{$host}{$port}为url数组，即host:port对应的所有的url
my %loc;
#key为某sockobj，value为数组，value[0]为接收到的原始数据，value[1]为解析后的http header
my %content;

my $has_num = 0;
while (1) {
	if ($reload == 1) {
		&init();
	}
	my $urls = &get_urls(\@whose_url, $total_client_num - $has_num);
	$has_num = $total_client_num;
	my $host_name;
	if (defined $urls) {
		##查DNS cache
		foreach (@$urls) {
			#my ($net_loc) = /:\/\/(.*?)(?:\/|$)/;
			my $url = new URI::URL $_;
			push @{$loc{$url->host}{$url->port}}, $_;
		}
		$host_name = &get_ip_by_name(keys %loc);
	}
	#与url中所有的web server建立连接，如果有一次连接失败，则在本次抓取时将不在从该web server进行抓取
	#在进行错误处理时，需要一个负反馈机制，能反馈给URL分发进程比较好(后期完善时再考虑)

	#调用connect，并加入到epoll集合
	foreach my $host (keys %$host_name) {
		my $ports = $loc{$url->host};
		foreach my $port (keys %$ports) {
			if (!exists $name_sockobj{$host}{$port}) {
				#调用connect并加入epoll
				$name_sockobj{$host}{$port} = IO::Socket::INET->new(
					Proto		=>	"tcp",
					Type		=>	SOCK_STREAM,
					Blocking	=>	0,
				);
				my $sock = $name_sockobj{$host}{$port};
				epoll_ctl($epfd, EPOLL_CTL_ADD, fileno $sock, EPOLLIN | EPOLLOUT | EPOLLET) >= 0 
					or die "epoll_ctl: $!\n";
				my $internet_addr = inet_aton($host_name->{$host});
				my $paddr = sockaddr_in($port, $internet_addr);
				$! = 0;
				my $ret = $sock->connect($paddr);
				my $err = $!;
				if ($ret) {
					if ($err == 0){	#连接建立
						epoll_ctl($epfd, EPOLL_CTL_MOD, fileno $sock, EPOLLIN | EPOLLET) >= 0
							or die "epoll_ctl: $!\n";
						$sock->syswrite(&get_http_request_str($loc{$host}{$port}->[0]));
						#print $sock &get_http_request_str($loc{$host}{$port}->[0]);
					}
					elsif ($err != EINPROGRESS) {	#建立连接时出错
						$sock = $name_sockobj{$host}{$port} = undef;
						$has_num -= scalar @{$ports->{$port}};
						next;
					}
					$fd_sockobj{fileno $name_sockobj{$host}{$port}} = [$sock, $host, $port];
				}
				else {	#建立连接时出错
					$sock = $name_sockobj{$host}{$port} = undef;
					$has_num -= scalar @{$ports->{$port}};
					next;
				}
			}
			else {
				next if (!defined $name_sockobj{$host}{$port});
				#已有连接
				my $sock = $name_sockobj{$host}{$port};
				$sock->syswrite(&get_http_request_str($loc{$host}{$port}->[0]));
			}
		}
	}
	##非阻塞抓取网页操作
	#使用Epoll建立非阻塞的connect，100个client同时发出请求。
	###########################################
	my $events = epoll_wait($epfd, 10, 5000);
	die "epoll_wait: $!\n" if (!defined $events);
	next if (scalar @$events == 0);
	foreach my $event (@$events) {
		my $host = $fd_sockobj{$event->[0]}[1];
		my $port = $fd_sockobj{$event->[0]}[2];
		my $sock = $fd_sockobj{$event->[0]}[0];
		if ($event->[1] | EPOLLOUT) {
			my $opt = $sock->getsockopt(SOL_SOCKET, SO_ERROR);
			unless $opt die "getsockopt: $!\n";
			if ($opt == 0) {
				epoll_ctl($epfd, EPOLL_CTL_MOD, $event->[0], EPOLLIN | EPOLLET) >= 0
					or die "epoll_ctl: $!\n";
				$sock->syswrite(&get_http_request_str($loc{$host}{$port}->[0]));
			}
			else {
				$name_sockobj{$host}{$port} = undef;
				$has_num -= scalar @{$loc{$host}{$port}};
			}
		}
		elsif ($event->[1] | EPOLLIN) {
			####################
			#$reald_done_flag的作用是防止对方返回正确数据的后面包含脏数据
			######################
			my $read_done_flag = 0;
			while (1) {
				my $buf;
				my $len = $sock->sysread($buf, 1024);
				if (defined $len) {
					if ($read_done_flag == 1) {
						next;
					}
					if ($len == 0) {	#连接关闭
						$name_sockobj{$host}{$port} = undef;
						last;
					}
					else {	#连接未关闭且有数据
						if (!defined $content{$sock}) {	#还没有接收到任何数据
							$content{$sock} = [""];
						}
						my $ret = &parse_http_response($buf, $len, $content{$sock});
						if ($ret == 0) {	#解析完毕
							$read_done_flag = 1;
							#last;	#此处不能last，应设置一个标志，将缓冲区内内容读取完毕(读取后丢弃)后再退出
						}
						elsif ($ret == 1) {
							next;
						}
						elsif ($ret == 2) {
							##############################################
							###打印日志，并设置标志，以将缓冲区内数据丢弃
							##############################################
							$read_done_flag = 1;
						}
					}
				}
				else {
					if ($! == EAGAIN) {	#内核缓冲区没有数据了
						print "";
					}
					else {	#读取时出错了
						warn "sysread: $!\n";
					}
					last;
				}
			}
			shift @{$loc{$host}{$port}};
			$has_num--;
		}
	}
}

#第一个参数为$buf，为sysread读取到的数据，第二个参数为$len为$buf中数据的长度
#第三个参数为列表引用，用于返回解析结果，列表第一个值为原始数据，第二个为解析后的HTTP::Response对象,第三个为第一个值的长度
#该函数解析http头以及进行unzip和chunked合并操作
#返回0表示一个HTTP头部和内容全部解析完毕
#返回1表示数据不够解析的，需要更多数据
#返回2表示解析时出错了，接收到的数据不对,可能的原因之一是未完全实现http协议
sub parse_http_response {
	my $buf = shift;
	my $len = shift;
	my $content = shift;

	if (!defined $content->[0]) {
		$content->[0] = $buf;
		#$content->[2] = $len;
	}
	else {
		$content->[0] .= $buf;
		#$content->[2] += $len;
	}
	#处理http response 头部
	if (!defined $content->[1]) {
		my ($header, $c) = ($content->[0] =~ /^(.*?)$CRLFCRLF(.*?)$/s);
		if (!defined $header) {
			return 1;
		}
		$content->[1] = HTTP::Response->parse($header);
		if (!defined $content->[1]) {
			return 2;
		}
		if (!defined $c) {
			$content->[0] = "";
			#$content->[2] = 0;
		}
		else {
			$content->[0] = $c;
			#$content->[2] = length $c;
		}
		$content->[2] = "";
	}
	#处理http体部分,主要处理chunked和gzip
	if (defined $content->[1]->header("Transfer-Encoding") and $content->[1]->header("Transfer-Encoding") eq "chunked") {
		while (1) {
			if ($content->[0] =~ /$CRLF/s) {
				if ($content->[0] =~ /^[0-9a-fA-F]+?$CRLF/s) {
					my ($chunked_len, $chunked_data) = ($content->[0] =~ /^([0-9a-fA-F]+?)$CRLF(.*?)/s);
					my $dlen = hex $chunked_len;
					if ($dlen == 0) {
						last;
					}
					if ($dlen > length $chunked_data + length $CRLF) {
						return 1;
					}
					else {
						$content->[2] .= substr $chunked_data, 0, $dlen;
						$content->[0] = substr $chunked_data, $dlen + 2 - 1;
					}
				}
				else {
					return 2;
				}
			}
			elsif ($content->[0] =~ /^[0-9a-fA-F]+?/) {
				return 1;
			}
			else {
				return 2;
			}
		}
	}
	elsif (defined $content->[1]->header("Content-Length")) {
		my $ContentLength = $content->[1]->header("Content-Length");
		if (length $content->[0] < $ContentLength) {
			return 1;
		}
		$content->[2] = $content->[0];
	}
	else {
		return 2;
	}

	if (defined $content->[1]->header("Content-Encoding")) {
		####################################################
		#处理gzip,使用$content->[2]
		####################################################
		if ($content->[1]->header("Content-Encoding") eq "gzip") {
			my $out;
			gunzip \$content->[2] => \$out or return 2;
			$content->[1]->content($out);
		}
		#####################################################
		#处理zlib
		#####################################################
		elsif ($content->[1]->header("Content-Encoding") eq "deflate") {
			my $status;
			my $i;
			($i, $status) = inflateInit();
			if ($status != Z_OK) {
				return 2;
			}
			my $out;
			($out, $status) = $i->inflate($content->[2]);
			if ($status != Z_STREAM_END) {
				return 2;
			}
			$content->[1]->content($out);
		}
		#####################################################
		#其他未实现
		#####################################################
		else {
			return 2;
		}
	}
	else {
		$content->[1]->content($content->[2]);
	}
	#################################
	#后续需要包括更多的http内容的解析
	#################################
	return 0;
}

sub get_http_request_str {
	my $uri = shift;
	my $req = HTTP::Request->new("GET", $uri);
	$req->header(
		"User-Agent" => $user_agent, 
		"Accept" => $accept, 
		"Accept-Language" => "zh-cn,zh;",
		"Accept-Encoding" => "gzip,deflate",
		"Accept-Charset" => "gb2312,utf-8",
		"Keep-Alive" => "300",
		"Connection" => "keep-alive",
	);
	return $req->as_string;
}

#输入host数组，输出host对应ip的散列表
sub get_ip_by_name {
	my $names = shift;
	foreach (@$names) {
		print $dns_cache_sock "getip ".$_."\n";
	}
	my %host_name;
	while (<$dns_cache_sock>) {
		my @host = split;
		next if $host[1] eq '0';
		$host_name{$host[0]} = $host[1];
	}
	return \%host_name;
}

sub get_urls {
	my $host = shift;
	my $num = shift;
	if ($num <= 0) {
		return undef;
	}
	my @urls;
	if (!defined $host) {
		print $url_dist_sock "geturl $num\n";
		push @urls, <$url_dist_sock>;
	}
	else {
		my $per_count = $num / @$host;
		for (my $i = 0; $i < $num; $i++) {
			my $t_host = $host->[$i % scalar @$host]
			print $url_dist_sock "geturl $t_host 1\n";
			push @urls, <$url_dist_sock>;
		}
	}
	return \@urls;
}

sub init {
	&load_conf($conf_file);
	$total_client_num = $conf->{"client"}->{"amount"};
	$user_agent = $conf->{"client"}->{"user_agent"};
	$accept = $conf->{"client"}->{"accept"};

	srand time();
	&conn_self_server(\@url_dist_host, $url_dist_sock);
	&conn_self_server(\@page_collect_host, $page_collect_sock);
	&conn_self_server(\@dns_cache_host, $dns_cache_sock);
}

#连接URL分发进程、网页汇总进程、DNS cache进程
sub conn_self_server {
	my $host_list = shift;
	my $sock = shift;

	close $sock if !&exists_addr($sock, $host_list);

	while (scalar @$host_list and !&exists_addr($sock, $host_list)) {
		my $selected_server = rand(scalar @$host_list);

		socket($sock, PF_INET, SOCK_STREAM, getprotobyname('tcp'));
		my $internet_addr = inet_aton($host_list->[$selected_server]->[0]) 
			or die "Couldn't convert ".$host_list->[$selected_server]->[0]." into an Internet address: $!\n";
		my $paddr = sockaddr_in($host_list->[$selected_server]->[1], $internet_addr);

		if (connect($sock, $paddr)) {
			last;
		}
		warn "Couldn't connect to ".$host_list->[$selected_server]->[0].":".$host_list->[$selected_server]->[1].": $!\n"
		splice @$host_list, $selected_server, 1;
		close $sock;
		$sock = undef;
	}
	die "Couldn't connect all of the url dist servers!\n" if (!scalar @$host_list);
}

sub exists_addr {
	my $sock = shift;
	my $addr_list = shift;
	if (!defined $sock) {
		return 0;
	}
	my $hersockaddr    = getpeername($sock);
	my ($port, $iaddr) = sockaddr_in($hersockaddr);
	my $herhostname    = gethostbyaddr($iaddr, AF_INET);
	my $herstraddr     = inet_ntoa($iaddr);
	foreach (@$addr_list) {
		return 1 if $_->[0] eq $herstraddr and $_->[1] == $port;
	}
	return 0;
}

sub load_conf {
	$conf = XMLin(shift);
	#将配置文件中配置所有服务器信息分别读入到三个列表中，便于后面随机抽取
	my $server = $conf->{"servers"}{"url_dist"};
	@url_dist_hot = ();
	foreach (keys %$server) {
		push @url_dist_host, [$server->{$_}{"ip"}, $server->{$_}{"port"}];
	}
	$server = $conf->{"servers"}{"page_collect"};
	@page_collect_host = ();
	foreach (keys %$server) {
		push @page_collect_host, [$server->{$_}{"ip"}, $server->{$_}{"port"}];
	}
	$server = $conf->{"servers"}{"dns_cache"};
	@dns_cache_host = ();
	foreach (keys %$server) {
		push @dns_cache_host, [$server->{$_}{"ip"}, $server->{$_}{"port"}];
	}
}
